package cn.csg.ams.collect2.poptopics.controller;

import cn.csg.ams.collect2.poptopics.datahub.TupleRecordDataUtils;
import cn.csg.ams.collect2.poptopics.infrastructure.utils.HuaWeiAuthentication;
import cn.csg.ams.collect2.poptopics.infrastructure.utils.JsonUtil;
import cn.csg.ams.collect2.poptopics.service.MkMcTermAlarmLogService;
import cn.hutool.core.map.MapUtil;
import com.aliyun.datahub.client.model.TupleRecordData;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.sql.Array;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Future;

/**
 * @author: zhanggang
 * @ClassName: MkMcTermAlarmLogController
 * @date: 2023-11-15 16:31:54
 * @Description: 停复电告警数据消费日志(mk_mc_term_alarm_log)表控制层
 */
@RestController
@RequestMapping("/member/V3/mkMcTermAlarmLog")
public class MkMcTermAlarmLogController {

    private static Logger logger = LoggerFactory.getLogger(MkMcTermAlarmLogController.class);

    /**
     * 服务对象
     */
    @Autowired
    private MkMcTermAlarmLogService mkMcTermAlarmLogService;

    @Autowired
    private KafkaProducer<String, String> kafkaProducer;

    @Autowired
    private KafkaConsumer<String, String> kafkaConsumer;

    @Autowired
    private JdbcTemplate jdbcTemplate;


    @PostConstruct
    public void test() {
        // List<Map<String, Object>> list = jdbcTemplate.queryForList("select * from cj_fhqx");

        // // 认证
        // String principal = "syy";
        // String keytab = "E:\\code\\outage_judge\\src\\main\\resources\\krb5.conf";
        // String krb5conf = "E:\\code\\outage_judge\\src\\main\\resources\\user.keytab";
        //
        // try {
        //     HuaWeiAuthentication.doAuthentication(principal, keytab, krb5conf);
        // } catch (IOException e) {
        //     e.printStackTrace();
        // }
        //
        // System.out.println(keytab);
        // System.out.println(krb5conf);

        // 写入模拟数据
        // List<Map<String, Object>> list = jdbcTemplate.queryForList("select * from mk_mc_term_alarm_log_copy1 where alarm_code = 'E200003401'");
        //
        // List<ProducerRecord<String, String>> records = new ArrayList<>();
        // for (Map<String, Object> map : list) {
        //     String key = MapUtil.getStr(map, "alarm_id");
        //     String value = JsonUtil.toJson(map);
        //     System.out.println(key);
        //     System.out.println(value);
        //     ProducerRecord<String, String> producerRecord = new ProducerRecord<>("test", key, value);
        //     Future<RecordMetadata> send = kafkaProducer.send(producerRecord);
        // }

        // 生产数据
        ProducerRecord<String, String> producerRecord = new ProducerRecord<>("test", "aaa", "bbb");
        Future<RecordMetadata> send = kafkaProducer.send(producerRecord);
        System.out.println("发送数据成功");

        // 消费数据
        kafkaConsumer.subscribe(Arrays.asList("test"));
        while (true) {
            ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofSeconds(10));
            if (records.isEmpty()) {
                logger.info("暂无数据");
                continue;
            }

            for (ConsumerRecord<String, String> record : records) {
                // TupleRecordData data = TupleRecordDataUtils.kafkaToDatahubValue(record.value());
                // alarmHandle.handle(data, record.timestamp());
                int partition = record.partition();
                System.out.println("分区" + partition + "消费到数据 " + record.key() + ":" + record.value());
            }
        }
    }

}

